package cn.lgwen.candy.runtime.driver;

import cn.lgwen.candy.runtime.operator.BufferingCollector;
import cn.lgwen.candy.runtime.operator.Collector;
import cn.lgwen.candy.runtime.operator.Processor;
import cn.lgwen.candy.runtime.queue.MemoryBlockQueue;
import cn.lgwen.candy.runtime.queue.Queue;

import java.util.LinkedList;
import java.util.List;

/**
 * 2019/12/24
 * aven.wu
 * danxieai258@163.com
 */
public class ProcessorDriver<I, O> implements Driver {

    /**
     * 上游的队列
     */
    private Queue<I> upstreamQueue;
    /**
     * 发往多个下游队列
     */
    private List<Queue<O>> forwardQueues;

    private Processor<I, O> processor;

    private int parallel = 1;

    public ProcessorDriver(Queue<I> upstreamQueue, Processor<I, O> processor) {
        this.upstreamQueue = upstreamQueue;
        this.processor = processor;
        forwardQueues = new LinkedList<>();
    }

    public void registerForward(Queue<O> queue) {
        forwardQueues.add(queue);
    }

    public void setUpstreamQueue(Queue<I> queue) {
        upstreamQueue = queue;
    }

    public ProcessorDriver(Processor<I, O> processor) {
        this.processor = processor;
        forwardQueues = new LinkedList<>();
    }

    @Override
    public void run() {
        processor.checkpoint(checkpoint());
        I input = upstreamQueue.take();
        Collector<O> collector = new BufferingCollector<>();
        processor.process(input, collector);
        while (collector.hasNext() && !forwardQueues.isEmpty()) {
            forwardQueues.forEach(queue -> queue.add(collector.next()));
        }
    }

    @Override
    public void parallel(int parallel) {
        this.parallel = parallel;
    }

    @Override
    public int parallel() {
        return parallel;
    }

    @Override
    public void checkpoint(boolean checkpoint) {

    }

}
